<?php
namespace Lib;

use Closure;

/**
 * Kafka相关的文档：https://arnaud-lb.github.io/php-rdkafka/phpdoc/book.rdkafka.html
 * https://github.com/arnaud-lb/php-rdkafka#examples
 * @package Lib
 */
class Kafka {
    protected $brokers = '127.0.0.1';// 指定 broker 地址,多个地址用"," 分割
    //protected $brokers = '10.10.10.207';// 指定 broker 地址,多个地址用"," 分割
    public function __construct($brokers = '127.0.0.1') {
        $kafka_config = Config::ini('KAFKA');//查询配置文件中的参数
        if(empty($kafka_config['HOST_NAME'])){ //不存在配置参数则使用默认参数
            $this->brokers = $brokers;
        }else{
            $this->brokers = $kafka_config['HOST_NAME'];//使用配置参数
        }
    }

    /**
     * 把数据写入队列
     * @param array $data 数据
     * @param string $topic
     */
    public function producer(array $data,$topic = 'test'){
        $rk = new \RdKafka\Producer();
        $rk->addBrokers($this->brokers);
        //$topic = $rk->newTopic('behavior_'.$data['bhv_type']);
        $topic = $rk->newTopic($topic);
        $topic->produce(RD_KAFKA_PARTITION_UA, 0, \MongoDB\BSON\fromPHP($data));//将数据发送到kafka队列(数据格式转化为BSON)。
        //$topic->produce(RD_KAFKA_PARTITION_UA, 0, \MongoDB\BSON\fromPHP($data));//将数据发送到kafka队列(数据格式转化为BSON)。
    }

    /**
     *  订阅并消费指定主题的数据
     * @param String $topic 主题名称
     * @param mixed $callback 匿名回调函数，使用此函数来消费kafka数据
     * @throws \Exception
     */
    public function consumer($topic,$callback){
        //消费者 Consumer
        $conf = new \RdKafka\Conf();

        // Set the group id. This is required when storing offsets on the broker    设置组id。这对代理时需要存储补偿;
        $conf->set('group.id', 'myConsumerGroup');

        $rk = new \RdKafka\Consumer($conf);
        $rk->addBrokers($this->brokers);

        $topicConf = new \RdKafka\TopicConf();
        $topicConf->set('auto.commit.interval.ms', 100);

        // Set the offset store method to 'file'

        $topicConf->set('offset.store.method', 'file');
        $topicConf->set('offset.store.path', sys_get_temp_dir());

        // Alternatively, set the offset store method to 'broker'
        // $topicConf->set('offset.store.method', 'broker');

        // Set where to start consuming messages when there is no initial offset in
        // offset store or the desired offset is out of range.
        // 'smallest': start from the beginning

        $topicConf->set('auto.offset.reset', 'smallest');

        $topic = $rk->newTopic($topic, $topicConf);//订阅用户行为点击的"主题"

        // Start consuming partition 0

        $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

        //$user_behavior_Model = new App\Model\Recommander\user_behavior();
        while (true) {
            // 第一个参数是分区号
            // 第二个参数是超时时间
            $message = $topic->consume(0, 120*10000);
            switch ($message->err) {
                case RD_KAFKA_RESP_ERR_NO_ERROR:
                    if (empty($message) or empty($message->payload)) continue;
                    $data = \MongoDB\BSON\toPHP($message->payload,[ 'root' => 'array', 'document' => 'array' ]);//bson转换为PHP对象
                    //$data = $message->payload;
                    //print_r($message->payload);
                    echo $message->topic_name;echo ' ';echo $message->offset;echo PHP_EOL;
                    //todo 写日志
                    if ($callback instanceof Closure ){
                        $callback($data);////使用匿名函数处理处理kafka数据
                    }
                    //$user_behavior_Model->insertOne($data,[],new \MongoDB\Driver\WriteConcern(0,100));//用户行为数据全部写入总表
                    break;
                case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                    echo "No more messages; will wait for more\n";
                    break;
                case RD_KAFKA_RESP_ERR__TIMED_OUT:
                    echo "Timed out\n";
                    break;
                default:
                    throw new \Exception($message->errstr(), $message->err);
                    break;
            }
        }
    }
}